package org.redis.cluster.loadbalance.routing;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;

import org.redis.cluster.RedisCluster;
import org.redis.cluster.cache.CacheFactory;
import org.redis.cluster.cache.Cache;
import org.redis.cluster.extension.ExtensionLoader;
import org.redis.cluster.interceptor.Interceptor;
import org.redis.cluster.interceptor.Invocation;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.transport.RedisReply;

public class RoutingInterceptor implements Interceptor,Plugin {
	
	private final static String ROUTING_COMMANDS_CONFIG="routing_commands";
	
	private final static String DEFAULT_ROUTING_COMMANDS="DEL,EXPIREAT,PEXPIRE,PEXPIREAT,"
			+ "APPEND,BITCOUNT,BITOP,DECR,DECRBY,INCR,INCRBY,INCRBYFLOAT,MSET,MSETNX,SET,SETBIT,SETEX,SETNX,SETRANGE,"
			+ "HDEL,HINCRBY,HINCRBYFLOAT,HMSET,HSET,HSETNX,LPOP,LPUSH,LPUSHX,LREM,LSET,LTRIM,RPOP,RPOPLPUSH,RPUSH,RPUSHX,"
			+ "SADD,SINTERSTORE,SMOVE,SPOP,SRANDMEMBER,SREM,"
			+ "ZADD,ZINCRBY,ZREM,ZREMRANGEBYRANK,ZREMRANGEBYSCORE";
	
	private final static String COMMAND_SPLITOR=",";
	
	private final static String EXPIRE_CONFIG="route_expire";
	
	private final static int DEFAULT_EXPIRE=1000*60;
	
	private RedisCluster redisCluster;
	private ExtensionLoader extensionLoader;
	private Cache cache;
	
	private boolean isEnable;
	private int expire;
	private Collection<String> commands;
	
	@Override
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.extensionLoader=redisCluster.getExtensionLoader();
		this.expire=initExpire();
		this.cache=initCache();
		this.isEnable=initEnable();
		this.commands=initCommands();
	}
	
	protected Cache initCache() {
		return CacheFactory.getCache();
	}
	
	protected int initExpire() {
		return redisCluster.getConfigManager().getSystemConfig().get(EXPIRE_CONFIG).getValue(Integer.class, DEFAULT_EXPIRE);
	}
	
	protected boolean initEnable() {
		return RoutingHelper.getEnable(redisCluster);
	}
	
	protected Collection<String> initCommands() {
		String config=redisCluster.getConfigManager().getSystemConfig().get(ROUTING_COMMANDS_CONFIG).getValue(String.class, DEFAULT_ROUTING_COMMANDS);
		return new HashSet<String>(Arrays.asList(config.split(COMMAND_SPLITOR)));
	}
	
	@Override
	public void destroy() {
	}

	@Override
	public RedisReply intercept(Invocation invocation) throws Throwable {
		if(!isEnable) {
			return invocation.invoke();
		}
		
		RedisCommand command=invocation.getCommand();
		if(!isRoutingCommand(command)) {
			return invocation.invoke();
		}

		doSetRouting(command);
		return invocation.invoke();
	}
	
	protected boolean isRoutingCommand(RedisCommand command) {
		if(!(command.getParameters().length>1)) {
			return false;
		}
		
		return commands.contains(command.getCommand());
	}
	
	protected boolean doSetRouting(RedisCommand command) {
		return cache.put(command.getParameters()[1], true, expire);
	}
	
 	public void setRedisCluster(RedisCluster redisCluster) {
		this.redisCluster = redisCluster;
	}
	public RedisCluster getRedisCluster() {
		return redisCluster;
	}
	public void setCache(Cache cache) {
		this.cache = cache;
	}
	public Cache getCache() {
		return cache;
	}
	public void setCommands(Collection<String> commands) {
		this.commands = commands;
	}
	public Collection<String> getCommands() {
		return commands;
	}
}